package com.allenkerr.rpc.registry;

import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;

/**
 * 功能：server 向 zookeeper 注册节点
 * 服务注册，zookeeper在该架构中扮演的是“服务注册表”的角色，
 * 用于注册所有服务器的地址与端口，并对客户端提供服务发现的功能。
 * <p>
 * Created by 小柯(Allen Kerr) on 17.5.23 18:41.
 */
public class ServiceRegistry {

    private static final Logger logger = LoggerFactory.getLogger(ServiceRegistry.class);

    private CountDownLatch latch = new CountDownLatch(1);

    /**
     * 注册地址，也即zookeeper的地址
     */
    private String registryAddress;

    /**
     * 通过构造函数传入zookeeper的地址
     * @param registryAddress
     */
    public ServiceRegistry(String registryAddress) {
        this.registryAddress = registryAddress;
    }

    /**
     * 创建zookeeper的连接
     *
     * @param data
     */
    public void register(String data) {
        if (data != null) {
            ZooKeeper zooKeeper = connectServer();
            if (zooKeeper != null) {
                createNode(zooKeeper, data);
            }
        }
    }

    /**
     * 连接到zookeeper，并创建监听
     *
     * @return
     */
    private ZooKeeper connectServer() {
        ZooKeeper zooKeeper = null;
        try {
            zooKeeper = new ZooKeeper(registryAddress, RpcConstant.ZK_SESSION_TIMEOUT,
                    new Watcher() {
                        @Override
                        public void process(WatchedEvent watchedEvent) {
                            if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                                latch.countDown();
                            }
                        }
                    });
            latch.await();
        } catch (Exception e) {
            logger.error("", e);
        }
        return zooKeeper;
    }

    /**
     * 创建zookeeper节点
     *
     * @param zooKeeper
     * @param data
     */
    private void createNode(ZooKeeper zooKeeper, String data) {
        byte[] bytes = data.getBytes();
        try {
            if (zooKeeper.exists(RpcConstant.ZK_REGISTRY_PATH, null) == null) {
                zooKeeper.create(RpcConstant.ZK_REGISTRY_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
            }
            String path = zooKeeper.create(RpcConstant.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);
            logger.debug("Create Zookeeper Node ({} => {})", path, data);
        } catch (Exception e) {
            logger.error("", e);
        }
    }
}
